package main

import (
	"demo/Conf"
	"fmt"
	kafka_go "gitee.com/tym_hmm/kafka-go"
	"time"
)

var (
	consumerFactory = kafka_go.NewFactoryConsumer()
	UTFALL_SECOND   = "2006-01-02 15:04:05"
	cstZone         = time.FixedZone("CST", 8*3600)
)

func main() {
	demoConsumerBuilder := kafka_go.NewBuildConsumer(Conf.Addr, Conf.GroupId3, Conf.Topic3)
	demoConsumerBuilder.SetDebug(true)
	demoConsumerBuilder.SetMultiplePartition(true)
	demoConsumerBuilder.SetIsAutoCommit(false)
	demoConsumerBuilder.SetKafkaVersion("3.0.0")
	demoConsumerBuilder.SetResponseListener(func(context *kafka_go.ConsumerMessageContext) {
		nowTime := time.Now().In(cstZone).Format(UTFALL_SECOND)
		fmt.Printf("%s xx received [partition:%d, topic:%s, content:%s]\n", nowTime, context.GetPartition(), context.GetTopic(), context.GetMessageString())
		//fmt.Printf("%+v\n", context.GetMessageString())
		//context.GetSession().Ack()
	})

	demo2ConsumerBuilder := kafka_go.NewBuildConsumer(Conf.Addr, Conf.GroupId2, Conf.Topic)
	demo2ConsumerBuilder.SetDebug(true)
	demo2ConsumerBuilder.SetMultiplePartition(false)
	demo2ConsumerBuilder.SetIsAutoCommit(false)
	demo2ConsumerBuilder.SetKafkaVersion("3.0.0")
	demo2ConsumerBuilder.SetResponseListener(func(context *kafka_go.ConsumerMessageContext) {
		nowTime := time.Now().In(cstZone).Format(UTFALL_SECOND)
		fmt.Printf("%s xx received2 [partition:%d, topic:%s, content:%s]\n", nowTime, context.GetPartition(), context.GetTopic(), context.GetMessageString())
		context.GetSession().Ack()
	})

	consumerFactory.RegisterConsumer(demo2ConsumerBuilder, demoConsumerBuilder)
	consumerFactory.Run()
}
